package com.member.mp.job;

import lombok.extern.slf4j.Slf4j;
import org.junit.Test;
import org.redisson.Redisson;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;

@Slf4j
public class DelayedQueueTest {

    @Test
    public void test() throws InterruptedException {
        // 配置并创建 Redisson 客户端
        Config config = new Config();
        config.useSingleServer().setAddress("redis://127.0.0.1:6379"); // 根据实际情况修改地址和端口
        RedissonClient redisson = Redisson.create(config);

        // 获取 RBlockingQueue 实例
        RBlockingQueue<String> blockingQueue = redisson.getBlockingQueue("orderQueue");

        // 获取 RDelayedQueue 实例
        RDelayedQueue<String> delayedQueue = redisson.getDelayedQueue(blockingQueue);

        // 示例 payId
        String payId = "12345";

        // 将数据存入延时队列
        String key = "ORDER_" + payId;
        delayedQueue.offer(key, 10, java.util.concurrent.TimeUnit.SECONDS); // 延迟1分钟
        System.out.println("Offered into delayed queue: " + key);

        // 模拟消费者从阻塞队列中获取数据
        new Thread(() -> {
            try {
                while (true) {
                    // 这个调用会阻塞直到有元素可取
                    String polledKey = blockingQueue.take();
                    System.out.println("Polled from blocking queue: " + polledKey);

                    // 处理任务逻辑
                    if (polledKey.startsWith("ORDER_")) {
                        System.out.println("Processing order: " + polledKey);
                    }
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                System.err.println("Thread was interrupted");
            }
        }).start();

        // 关闭 Redisson 客户端前需要先关闭 RDelayedQueue
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            redisson.shutdown(); // 关闭 Redisson 客户端
        }));

        // 等待一段时间以便观察结果
        Thread.sleep(2 * 60 * 1000); // 等待2分钟以确保延迟任务被执行
    }
}